package pt.inescid.components.communicationservice;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import pt.inescid.components.api.CommunicationServiceI;
import pt.inescid.components.api.Component;
import pt.inescid.components.messages.ACKMsg;
import pt.inescid.components.messages.AvailabilityMsg;
import pt.inescid.components.messages.Task;
import pt.inescid.components.messages.TaskResult;
import pt.inescid.components.messages.UpdateMsg;
import pt.inescid.core.GridP2P;
import pt.inescid.utils.Log;
import rice.p2p.commonapi.Application;
import rice.p2p.commonapi.Endpoint;
import rice.p2p.commonapi.Id;
import rice.p2p.commonapi.Message;
import rice.p2p.commonapi.NodeHandle;
import rice.p2p.commonapi.RouteMessage;
import rice.p2p.commonapi.rawserialization.InputBuffer;
import rice.p2p.commonapi.rawserialization.MessageDeserializer;
import rice.p2p.commonapi.rawserialization.RawMessage;
import rice.pastry.PastryNode;

/**
 * Communication Service
 * 
 * @author Filipe Paredes
 * @author Sergio Esteves
 */
public class CommunicationService extends Component implements Application, Runnable, CommunicationServiceI {

	protected final static int nThreadsToRcvMsgs = 30;

	// List of the node Ids that requested the use of threads to the pool for
	// receiving messages from the network (another similar list exists in 
	// GiGiSimulator for sending messages).
	List<Id> threadedNodes = new ArrayList<Id>();
	/**
	 * The Endpoint represents the underlying node.  By making calls on the 
	 * Endpoint, it assures that the message will be delivered to a Communication
	 * Service on whichever node the message is intended for.
	 */
	protected Endpoint endpoint;
	protected PastryNode node;
	
	private ExecutorService pool;
	//private List updateThreadFuture;
	private List<paramsMessage> waitingToReceive;
	private List<Id> joiningNodes;

	// With Future we can wait for a thread launched in 'deliver' method to end.
	//private Future receiverFuture;
	public CommunicationService(GridP2P gridP2P) {
		super(gridP2P);		
		this.pool = Executors.newFixedThreadPool(nThreadsToRcvMsgs);
		//this.updateThreadFuture = new ArrayList();
		this.waitingToReceive = new ArrayList<paramsMessage>();
		this.joiningNodes = new ArrayList<Id>();
		
		log.write(this, "CommunicationService constructed", Log.FINER);
	}

	public void registerEndPoint() {		
		node = gridP2P.getNode();
		
		endpoint = node.buildEndpoint(this, "myinstance");
		endpoint.setDeserializer(new MessageDeserializer() {

			public Message deserialize(InputBuffer buf, short type, int priority,
					NodeHandle sender) throws IOException {
				switch (type) {
				case Task.TYPE:
					return new Task(buf, endpoint);
				case UpdateMsg.TYPE:
					return new UpdateMsg(buf, endpoint);
				case TaskResult.TYPE:
					return new TaskResult(buf, endpoint);
				case ACKMsg.TYPE:
					return new ACKMsg(buf, endpoint);				
				case AvailabilityMsg.TYPE:
					return new AvailabilityMsg(buf, endpoint);
				}
				return null;
			}
		});
		// now we can receive messages
		endpoint.register();
		
		log.write(this, "CommunicationService initialized", Log.SEVERE);
	}	

	/**
	 * Getter for the endpoint
	 */
	public Endpoint getEndpoint() {
		return endpoint;
	}

	/**
	 * Getter for the node Id.
	 */
	public Id getNodeId() {
		return endpoint.getId();
	}

	/**
	 * Called to route a message to the id
	 */
	public void routeMyMsg(Id id, Message message) {
		log.write(this, "routeMyMsg called (Id: " + id + ", message: " +
				message + ")", Log.FINER);
		
		endpoint.route(id, (RawMessage) message, null);
	}

	/**
	 * Called to directly send a message to the nh
	 */
	public void routeMyMsgDirect(Message msg, NodeHandle nh) {
		log.write(this, "routeMyMsgDirect called (message: " + msg + 
				", nodeHandle: " + nh + ")", Log.FINER);
		endpoint.route(null, msg, nh);
	}

	/**
	 * Called when we receive a message. The pool.execute runs a new Thread
	 */
	public void deliver(Id id, Message message) {
		log.write(this, "deliver called - Message received (id: " + id + ", message: " +
				message + ")", Log.FINER);
		
		long currentTime = endpoint.getEnvironment().getTimeSource().currentTimeMillis();
		synchronized (this.waitingToReceive) {
			waitingToReceive.add(waitingToReceive.size(), new paramsMessage(id, message, currentTime));
		}
		pool.execute(this);	
	}

	public void receive(Id id, Message message, long currentTime) {
		log.write(this, "receive called (id: " + id + ", message: " + 
				message + ", currentTime: " + currentTime + ")", Log.FINER);
		
		switch (((RawMessage) message).getType()) {
		case Task.TYPE: {
			log.write(this,"message received (type: Task, Id: " + id + ", Message: " + 
					message.hashCode() + ", currentTime: " + currentTime + ")", Log.FINE);
					
//			printInfo(String.valueOf(message.hashCode()),
//					this + " received Gridlet " + message.hashCode() + " from " + id +
//					" (real time = " + System.currentTimeMillis() + ")", null);

			gridP2P.messageReceived(id, (Task)message, currentTime);			
			break;
		}
		case UpdateMsg.TYPE: {
			log.write(this,"message received (type: UpdateMsg, Id: " + id + ", Message: " + 
					message.hashCode() + ", currentTime: " + currentTime + ")", Log.FINER);
			gridP2P.messageReceived(((UpdateMsg)message).getSource() , (UpdateMsg)message);

			// if this was the first update message from a new joined node...
			synchronized (joiningNodes) {
				if (joiningNodes.remove(id))
					routeMyMsg(id, gridP2P.buildUpdateMsg());
			}
			break;
		}
		case TaskResult.TYPE: {
			log.write(this,"message received (type: TaskResult, Id: " + id + ", Message: " + 
					message.hashCode() + ", currentTime: " + currentTime + ")", Log.FINER);
			
			gridP2P.messageReceived(id, (TaskResult)message, currentTime);
			break;
		}
		case ACKMsg.TYPE: {
			log.write(this,"message received (type: ACKMsg, Id: " + id + ", Message: " + 
					message.hashCode() + ")", Log.FINER);
			gridP2P.messageReceived(id, (ACKMsg)message);
			break;
		}
		case AvailabilityMsg.TYPE: {
			log.write(this,"AvailabilityMsg received (Id: " + id + ", Message: " + 
					message.hashCode() + ")", Log.FINER);
			gridP2P.messageReceived(id, (AvailabilityMsg)message);			
			break;
		}
		}
	}

	/**
	 * Called when a message is received
	 */ 
	public void deliverBroadcast(Id id, Message message) {
		log.write(this, "deliverBroadcast called (id: " + id + ", message: " + message + ")", Log.FINER);
		//System.out.println("Message ;" + this.getNode().getId() + message);
	}

	/**
	 * Called when we receive a message.
	 */
	public void forwardBroadcast(Id id, Message message) {
		log.write(this, "forwardBroadcast called (id: " + id + ", message: " + message + ")", Log.FINER);
		//System.out.println("Forward ;" + this.getNode().getId() + message);
	}

	/**
	 * Called when you hear about a new neighbor
	 */
	public void update(NodeHandle handle, boolean joined) {
		log.write(this, "update called (handle: " + handle + 
				", joined: " + joined + ")", Log.FINER);			   
		
		// network initialization completed?
		if(!gridP2P.networkInitialized())
			return;

		if (joined) {
			gridP2P.joinNode(handle);			
		} else {
			gridP2P.leaveNode(handle);
		}
	}

	/**
	 * Called a message travels along your path.
	 * Don't worry about this method for now.
	 */
	public boolean forward(RouteMessage message) {
		return true;
	}

	public String toString() {
		return "Node " + endpoint.getId();
	}

	public void run() {
		Id id;
		Message message;
		long currentTime;
		
		synchronized (waitingToReceive) {
			paramsMessage pm = waitingToReceive.get(0);
			id = pm.id;
			message = pm.message;
			currentTime = pm.currentTime;
			waitingToReceive.remove(0);
		}
		
		receive(id, message, currentTime);
	}
}

class paramsMessage {

	public Id id;
	public Message message;
	public long currentTime;

	public paramsMessage(Id id, Message message, long currentTime) {
		this.id = id;
		this.message = message;
		this.currentTime = currentTime;
	}
}
